package com.sinozo.data.app.dws;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sinozo.data.app.function.ActiveUserTimestampsAndWatermarks;
import com.sinozo.data.bean.TrafficActiveUserBean;
import com.sinozo.data.bean.key.NewUserKeyByBean;
import com.sinozo.data.common.ConfigConstant;
import com.sinozo.data.common.DataConfig;
import com.sinozo.data.utils.*;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @author zhaoyb
 * @version 1.0
 * @description 活跃用户
 * @date 2023/5/25 18:23:47
 */
public class DwsTrafficActiveUserWindow {

    public static void main(String[] args) {
        Logger logger = LoggerFactory.getLogger(DwsTrafficActiveUserWindow.class);
        try {
            //1.获取执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(DataConfig.KAFKA_PARTITION_NUM);

            //1.1 开启CheckPoint
            env.enableCheckpointing(DataConfig.FLINK_CHECKPOINT_INTERVAL, CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(DataConfig.FLINK_CHECKPOINT_TIMEOUT);
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);

            //1.2 设置状态后端
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
            env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
            );
            env.setRestartStrategy(RestartStrategies.failureRateRestart(10, org.apache.flink.api.common.time.Time.of(3L, TimeUnit.DAYS), org.apache.flink.api.common.time.Time.of(1L, TimeUnit.MINUTES)));

            String hdfsPath = DataConfig.HDFS_SERVER;
            env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(PathUtil.getHdfsPath(hdfsPath,"dwsTrafficActiveUserWindow")));
            env.setStateBackend(new EmbeddedRocksDBStateBackend());

            // 2.读取三个主题的数据创建流
            String uvTopic = ConfigConstant.KAFKA_UNIQUE_START_TOPIC;
            String startTopic = ConfigConstant.KAFKA_START_LOG_TOPIC;
            String onlineTopic = ConfigConstant.KAFKA_ONLINE_LOG_TOPIC;

            String groupId = ConfigConstant.KAFKA_ACTIVE_USER_START_DETAIL_GROUP_ID;

            DataStreamSource<String> uvDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(uvTopic, groupId));
            DataStreamSource<String> startDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(startTopic, groupId));
            DataStreamSource<String> onlineDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(onlineTopic, groupId));

            // 3.统一数据格式
            SingleOutputStreamOperator<TrafficActiveUserBean> userUniqueWithUvDS = uvDS.map(line -> {
                JSONObject jsonObject = JSON.parseObject(line);
                JSONObject common = jsonObject.getJSONObject("common");
                Long ts = jsonObject.getLong("ts");

                TrafficActiveUserBean bean = TrafficActiveUserBean.builder().appVer(common.getString("app_ver")).os(common.getString("os"))
                        .qid(common.getString("qid")).groupQid(ValueEmptyUtils.emptyString(common.getString("group_qid")))
                        .ascQid(common.getString("asc_qid")).code(common.getString("code"))
                        .labCode(common.getString("lab_code")).labGroupCode(common.getString("lab_group_code"))
                        .dau(1L).startCnt(0L).coldStartCnt(0L).hotStartCnt(0L).useTime(0L).ts(ts).build();
                return bean;
            });

            SingleOutputStreamOperator<TrafficActiveUserBean> startWithPageDS = startDS.map(line -> {
                JSONObject jsonObject = JSON.parseObject(line);
                JSONObject common = jsonObject.getJSONObject("common");
                Long ts = jsonObject.getLong("ts");
                Integer startWay = jsonObject.getJSONObject("start").getInteger("start_way");

                TrafficActiveUserBean bean = TrafficActiveUserBean.builder().appVer(common.getString("app_ver")).os(common.getString("os"))
                        .qid(common.getString("qid")).groupQid(ValueEmptyUtils.emptyString(common.getString("group_qid")))
                        .ascQid(common.getString("asc_qid")).code(common.getString("code"))
                        .labCode(common.getString("lab_code")).labGroupCode(common.getString("lab_group_code"))
                        .dau(0L).startCnt(1L).coldStartCnt((startWay == 2 || startWay == 1) ? 1L : 0L).hotStartCnt(startWay == 0 ? 1L : 0L)
                        .useTime(0L).ts(ts).build();
                return bean;
            });

            SingleOutputStreamOperator<TrafficActiveUserBean> onlinePageDS = onlineDS.map(line -> {
                JSONObject jsonObject = JSON.parseObject(line);
                JSONObject common = jsonObject.getJSONObject("common");
                Long ts = jsonObject.getLong("ts");
                Long onlineTime = jsonObject.getJSONObject("online").getLong("online_time");

                TrafficActiveUserBean bean = TrafficActiveUserBean.builder().appVer(common.getString("app_ver")).os(common.getString("os"))
                        .qid(common.getString("qid")).groupQid(ValueEmptyUtils.emptyString(common.getString("group_qid")))
                        .ascQid(common.getString("asc_qid")).code(common.getString("code"))
                        .labCode(common.getString("lab_code")).labGroupCode(common.getString("lab_group_code"))
                        .dau(0L).startCnt(0L).coldStartCnt(0L).hotStartCnt(0L).useTime(onlineTime).ts(ts).build();
                return bean;
            });

            // 4.将流进行Union
            DataStream<TrafficActiveUserBean> unionDS = startWithPageDS.union(onlinePageDS, userUniqueWithUvDS);

            // 5.提取事件时间生成WaterMark
            SingleOutputStreamOperator<TrafficActiveUserBean> userInstallViewWithWmDS = unionDS.assignTimestampsAndWatermarks(new ActiveUserTimestampsAndWatermarks());

            // 6.分组开窗聚合
            WindowedStream<TrafficActiveUserBean, NewUserKeyByBean, TimeWindow> windowedStream = userInstallViewWithWmDS.keyBy(new KeySelector<TrafficActiveUserBean, NewUserKeyByBean>() {
                @Override
                public NewUserKeyByBean getKey(TrafficActiveUserBean value) throws Exception {
                    NewUserKeyByBean keyByBean = new NewUserKeyByBean();
                    BeanUtils.copyProperties(keyByBean, value);
                    return keyByBean;
                }
            }).window(TumblingEventTimeWindows.of(DataConfig.WINDOW_TIME));


            SingleOutputStreamOperator<TrafficActiveUserBean> resultDS = windowedStream.reduce(new ReduceFunction<TrafficActiveUserBean>() {
                @Override
                public TrafficActiveUserBean reduce(TrafficActiveUserBean value1, TrafficActiveUserBean value2) throws Exception {
                    value1.setDau(value1.getDau() + value2.getDau());
                    value1.setStartCnt(value1.getStartCnt() + value2.getStartCnt());
                    value1.setColdStartCnt(value1.getColdStartCnt() + value2.getColdStartCnt());
                    value1.setHotStartCnt(value1.getHotStartCnt() + value2.getHotStartCnt());
                    value1.setUseTime(value1.getUseTime() + value2.getUseTime());
                    return value1;
                }
            }, new WindowFunction<TrafficActiveUserBean, TrafficActiveUserBean, NewUserKeyByBean, TimeWindow>() {
                @Override
                public void apply(NewUserKeyByBean key, TimeWindow window, Iterable<TrafficActiveUserBean> input, Collector<TrafficActiveUserBean> out) throws Exception {
                    //获取数据
                    TrafficActiveUserBean next = input.iterator().next();
                    //补充信息
                    next.setWinStartTime(DateFormatUtil.toYmdHms(window.getStart()));
                    next.setWinEndTime(DateFormatUtil.toYmdHms(window.getEnd()));
                    //修改TS
                    next.setTs(System.currentTimeMillis());
                    next.setCreateTime(DateFormatUtil.toYmdHms(next.getTs()));
                    //输出数据
                    out.collect(next);
                }
            });

            resultDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_user_active_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

            // 8.启动任务
            env.execute("DwsTrafficActiveUserWindow");
        } catch (Exception e) {
            logger.error("error exec", e);
        }

    }
}
